

package rafthttp

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"path"
	"strings"
	"sync"
	"time"

	"go.etcd.io/etcd/api/v3/version"
	"go.etcd.io/etcd/client/pkg/v3/transport"
	"go.etcd.io/etcd/client/pkg/v3/types"
	"go.etcd.io/etcd/pkg/v3/httputil"
	"go.etcd.io/etcd/raft/v3/raftpb"
	stats "go.etcd.io/etcd/server/v3/etcdserver/api/v2stats"

	"github.com/coreos/go-semver/semver"
	"go.uber.org/zap"
	"golang.org/x/time/rate"
)

const (
	streamTypeMessage  streamType = "message"
	streamTypeMsgAppV2 streamType = "msgappv2"

	streamBufSize = 4096
)

var (
	errUnsupportedStreamType = fmt.Errorf("unsupported stream type")

	// 密钥为字符串格式“major.minor.patch”
	supportedStream = map[string][]streamType{
		"2.0.0": {},
		"2.1.0": {streamTypeMsgAppV2, streamTypeMessage},
		"2.2.0": {streamTypeMsgAppV2, streamTypeMessage},
		"2.3.0": {streamTypeMsgAppV2, streamTypeMessage},
		"3.0.0": {streamTypeMsgAppV2, streamTypeMessage},
		"3.1.0": {streamTypeMsgAppV2, streamTypeMessage},
		"3.2.0": {streamTypeMsgAppV2, streamTypeMessage},
		"3.3.0": {streamTypeMsgAppV2, streamTypeMessage},
		"3.4.0": {streamTypeMsgAppV2, streamTypeMessage},
		"3.5.0": {streamTypeMsgAppV2, streamTypeMessage},
		"3.6.0": {streamTypeMsgAppV2, streamTypeMessage},
	}
)

type streamType string

func (t streamType) endpoint(lg *zap.Logger) string {
	switch t {
	case streamTypeMsgAppV2:
		return path.Join(RaftStreamPrefix, "msgapp")
	case streamTypeMessage:
		return path.Join(RaftStreamPrefix, "message")
	default:
		if lg != nil {
			lg.Panic("unhandled stream type", zap.String("stream-type", t.String()))
		}
		return ""
	}
}

func (t streamType) String() string {
	switch t {
	case streamTypeMsgAppV2:
		return "stream MsgApp v2"
	case streamTypeMessage:
		return "stream Message"
	default:
		return "unknown stream"
	}
}

var (
	// linkHeartbeatMessage是在
	// 链接层中用作心跳消息的特殊消息。它从不与raft的消息冲突，因为raft 
	// 在没有from和To字段的情况下不会发送消息。
	linkHeartbeatMessage = raftpb.Message{Type: raftpb.MsgHeartbeat}
)

func isLinkHeartbeatMessage(m *raftpb.Message) bool {
	return m.Type == raftpb.MsgHeartbeat && m.From == 0 && m.To == 0
}

type outgoingConn struct {
	t streamType
	io.Writer
	http.Flusher
	io.Closer

	localID types.ID
	peerID  types.ID
}

// 向Stream消息通道写入消息
type streamWriter struct {
	lg *zap.Logger
	localID types.ID
	peerID  types.ID	// 对端节点的ID。
	status *peerStatus
	fs     *stats.FollowerStats
	r      Raft	// 底层的Raft实例。
	mu      sync.Mutex // 保护字段工作和关闭
	closer  io.Closer	// 负责关闭底层的长连接。
	working bool	// 负责标识当前的streamWriter是否可用（底层是否关联了相应的网络连接）。
	// 通过前面对Peer的分析可知，Peer会将待发送的消息写入该通道，streamWriter则从该通道中读取消息并发送出去。
	msgc  chan raftpb.Message
	// 通过该通道获取当前streamWriter实例关联的底层网络连接，outgoingConn其实是对网络连接的一层封装，
	// 其中记录了当前连接使用的协议版本，以及用于关闭连接的Flusher和Closer等信息。
	connc chan *outgoingConn
	stopc chan struct{}
	done  chan struct{}
}

// 启动StreamWriter创建streamWrite并启动长时间运行的go例程，该例程接受
// 消息并写入连接的传出连接。
func startStreamWriter(lg *zap.Logger, local, id types.ID, status *peerStatus,
	fs *stats.FollowerStats, r Raft) *streamWriter {
	w := &streamWriter{
		lg: lg,

		localID: local,
		peerID:  id,

		status: status,
		fs:     fs,
		r:      r,
		msgc:   make(chan raftpb.Message, streamBufSize),
		connc:  make(chan *outgoingConn),
		stopc:  make(chan struct{}),
		done:   make(chan struct{}),
	}
	go w.run()
	return w
}

// peer.start（）方法中是通过调用startStreamWriter（）方法初始化上述字段并启动streamWriter 实例的，
// 其中还启动了一个后台goroutine来执行streamWriter.run（）方法。
// 在streamWriter.run() 方法中，主要完成了下面三件事情：
// (1）当其他节点主动与当前节点创建连接（即Stream消息通道底层使用的网络连接〉时，
// 该连接实例会写入对应peer.writer.connc通道，在streamWriter.run（）方法中会通过该通道获取该连接实例井进行绑定，
// 之后才能开始后续的消息发送。
// (2）定时发送心跳消息，该心跳消息并不是前面介绍etcd-raft模块时提到的MsgHeartbeat消息，
// 而是为了防止底层连接超时的消息，后面会详细介绍该消息的处理过程。
// (3）发送除心跳消息外的其他类型的消息。
func (cw *streamWriter) run() {
	var (
		msgc       chan raftpb.Message	// 指向当前streamWriter.msgc字段
		// 定时器会定时向该通道发送信号，触发心跳消息的发送，该心跳消息的主要目的是为了防止连接长时间不用断开的
		heartbeatc <-chan time.Time
		t          streamType	// 用来记录消息的版本信息
		enc        encoder	// 编码器，负责将消息序列化并写入迄接的缓冲区
		flusher    http.Flusher	// 负责刷新底层连接，将数据真正发送出去
		batched    int	// 当前未Flush的消息个数
	)
	tickc := time.NewTicker(ConnReadTimeout / 3)	// 发送心跳消息的定时器
	defer tickc.Stop()
	unflushed := 0	// 未Flush的字节数

	if cw.lg != nil {
		cw.lg.Info(
			"started stream writer with remote peer",
			zap.String("local-member-id", cw.localID.String()),
			zap.String("remote-peer-id", cw.peerID.String()),
		)
	}

	for {
		select {
		case <-heartbeatc:	// 定时器到期，触发心跳消息
			// 通过encoder编码心跳消息
			err := enc.encode(&linkHeartbeatMessage)
			unflushed += linkHeartbeatMessage.Size()	// 增加未Flush出去的字节数
			// 若没有异常，则使用flusher~＋缓存的消息全部发送出去，并重置batched和unflushed两个统计交量
			if err == nil {
				flusher.Flush()
				batched = 0
				sentBytes.WithLabelValues(cw.peerID.String()).Add(float64(unflushed))
				unflushed = 0
				continue
			}

			cw.status.deactivate(failureType{source: t.String(), action: "heartbeat"}, err.Error())

			sentFailures.WithLabelValues(cw.peerID.String()).Inc()
			cw.close()	// 若发生异常，则关闭streamWriter，会导致底层连接的关闭
			if cw.lg != nil {
				cw.lg.Warn(
					"lost TCP streaming connection with remote peer",
					zap.String("stream-writer-type", t.String()),
					zap.String("local-member-id", cw.localID.String()),
					zap.String("remote-peer-id", cw.peerID.String()),
				)
			}
			// 将heartbeatc和msgc两个通道清空，后续就不会再发送心跳消息和其他类型的消息了
			heartbeatc, msgc = nil, nil

		case m := <-msgc:	// peer向streamWriter.msgc写入待发送的消息
			err := enc.encode(&m)	// 将消息序列化并写入底层连接
			if err == nil {	// 若没有异常，则逆士曾unflushed变量
				unflushed += m.Size()
				// msgc通道中的消息全部发送完成或是未Flush的消息较多，则触发Flush,否则只是递增batched变量
				if len(msgc) == 0 || batched > streamBufSize/2 {
					flusher.Flush()
					sentBytes.WithLabelValues(cw.peerID.String()).Add(float64(unflushed))
					unflushed = 0
					batched = 0
				} else {
					batched++
				}

				continue
			}

			cw.status.deactivate(failureType{source: t.String(), action: "write"}, err.Error())
			cw.close()
			if cw.lg != nil {
				cw.lg.Warn(
					"lost TCP streaming connection with remote peer",
					zap.String("stream-writer-type", t.String()),
					zap.String("local-member-id", cw.localID.String()),
					zap.String("remote-peer-id", cw.peerID.String()),
				)
			}
			// 若发生异常，则关闭streamWriter，清空heartbeatc和msgc两个通道，
			heartbeatc, msgc = nil, nil
			cw.r.ReportUnreachable(m.To)
			sentFailures.WithLabelValues(cw.peerID.String()).Inc()

		// 当其他节点主动与当前节点创建Stream消息通道时，会先通过StreamHandler的处理，
		// StreamHandler会通过attach（）方法将连接写入对应peer.writer.connc通道，
		// 而当前的goroutine会通过该通道获取连接，然后开始发送消息
		case conn := <-cw.connc:	// 获取与当前streamWriter实例绑定的底层连接
			cw.mu.Lock()
			closed := cw.closeUnlocked()
			t = conn.t	// 获取该连接底层发送的消息版本，并创建相应的encoder实例
			switch conn.t {
			case streamTypeMsgAppV2:
				enc = newMsgAppV2Encoder(conn.Writer, cw.fs)
			case streamTypeMessage:	// 这里只关注最新版本，忽略其他版本
				// 将http.ResponseWriter封装成messageEncoder，上层调用通过messageEncoder实例完成消息发送
				enc = &messageEncoder{w: conn.Writer}
			default:
				if cw.lg != nil {
					cw.lg.Panic("unhandled stream type", zap.String("stream-type", t.String()))
				}
			}
			if cw.lg != nil {
				cw.lg.Info(
					"set message encoder",
					zap.String("from", conn.localID.String()),
					zap.String("to", conn.peerID.String()),
					zap.String("stream-type", t.String()),
				)
			}
			flusher = conn.Flusher	// 记录底层连接对应的Flusher
			unflushed = 0	// 重直未Flush的字节数
			cw.status.activate()	// peerStatus.active设置为true
			cw.closer = conn.Closer	// 记录底层连接对应的Flusher
			cw.working = true	// 标识当前streamWriter正在运行
			cw.mu.Unlock()

			if closed {
				if cw.lg != nil {
					cw.lg.Warn(
						"closed TCP streaming connection with remote peer",
						zap.String("stream-writer-type", t.String()),
						zap.String("local-member-id", cw.localID.String()),
						zap.String("remote-peer-id", cw.peerID.String()),
					)
				}
			}
			if cw.lg != nil {
				cw.lg.Info(
					"established TCP streaming connection with remote peer",
					zap.String("stream-writer-type", t.String()),
					zap.String("local-member-id", cw.localID.String()),
					zap.String("remote-peer-id", cw.peerID.String()),
				)
			}
			// 更新heartbeatc和msgc两个远远，自此之后，才能发送消息
			heartbeatc, msgc = tickc.C, cw.msgc

		case <-cw.stopc:
			if cw.close() {
				if cw.lg != nil {
					cw.lg.Warn(
						"closed TCP streaming connection with remote peer",
						zap.String("stream-writer-type", t.String()),
						zap.String("remote-peer-id", cw.peerID.String()),
					)
				}
			}
			if cw.lg != nil {
				cw.lg.Info(
					"stopped TCP streaming connection with remote peer",
					zap.String("stream-writer-type", t.String()),
					zap.String("remote-peer-id", cw.peerID.String()),
				)
			}
			close(cw.done)
			return
		}
	}
}

func (cw *streamWriter) writec() (chan<- raftpb.Message, bool) {
	cw.mu.Lock()
	defer cw.mu.Unlock()
	return cw.msgc, cw.working
}

func (cw *streamWriter) close() bool {
	cw.mu.Lock()
	defer cw.mu.Unlock()
	return cw.closeUnlocked()
}

func (cw *streamWriter) closeUnlocked() bool {
	if !cw.working {
		return false
	}
	if err := cw.closer.Close(); err != nil {
		if cw.lg != nil {
			cw.lg.Warn(
				"failed to close connection with remote peer",
				zap.String("remote-peer-id", cw.peerID.String()),
				zap.Error(err),
			)
		}
	}
	if len(cw.msgc) > 0 {
		cw.r.ReportUnreachable(uint64(cw.peerID))
	}
	cw.msgc = make(chan raftpb.Message, streamBufSize)
	cw.working = false
	return true
}
// attach（）方法，该方法会接收outgoingConn实例井写入streamWriter.connc通道中，
// peer.attachOutgoingConn（）方法就是通过调用该方法实现的。在streamHandler中，
// 也就是通过调用peer.attachOutgoingConn（）方法将底层网络连接传递到streamWriter中的。
func (cw *streamWriter) attach(conn *outgoingConn) bool {
	select {
	case cw.connc <- conn:
		return true
	case <-cw.done:
		return false
	}
}

func (cw *streamWriter) stop() {
	close(cw.stopc)
	<-cw.done
}

// streamReader是一个长时间运行的go例程，它可以拨号到远程流
// 端点并从返回的响应体读取消息。
type streamReader struct {
	lg *zap.Logger
	peerID types.ID	// 对应节点的ID。
	typ    streamType	// 关联的底层连接使用的协议版本信息。
	tr     *Transport	// 关联的rafthttp.Transport实例。
	picker *urlPicker	// 用于获取对端节点的可用的URL。
	status *peerStatus
	// 创建streamReader实例时是使用peer.reeve通道初始化该宇段的，其中还会启动一个后台goroutine从peer.reeve通道中读取消息。
	// 从对端节点发送来的非MsgProp类型的消息会首先由streamReader写入reeve通道中，
	// 然后由peer.start()启动的后台goroutine读取出来，交由底层的eted-raft模块进行处理。
	recvc  chan<- raftpb.Message
	// recvc通道类似，只不过其中接收的是MsgProp类型的消息。
	propc  chan<- raftpb.Message
	rl *rate.Limiter // 更改拨号重试尝试的频率
	errorc chan<- error
	mu     sync.Mutex
	paused bool	// 是否暂停读取数据。
	closer io.Closer
	ctx    context.Context
	cancel context.CancelFunc
	done   chan struct{}
}

func (cr *streamReader) start() {
	cr.done = make(chan struct{})
	if cr.errorc == nil {
		cr.errorc = cr.tr.ErrorC
	}
	if cr.ctx == nil {
		cr.ctx, cr.cancel = context.WithCancel(context.Background())
	}
	go cr.run()
}

func (cr *streamReader) run() {
	t := cr.typ	// 获取使用的消息版本

	if cr.lg != nil {
		cr.lg.Info(
			"started stream reader with remote peer",
			zap.String("stream-reader-type", t.String()),
			zap.String("local-member-id", cr.tr.ID.String()),
			zap.String("remote-peer-id", cr.peerID.String()),
		)
	}

	for {
		// 向对端节点发送一个GET请求，然后获取并返回相应的ReadCloser
		rc, err := cr.dial(t)
		if err != nil {
			if err != errUnsupportedStreamType {
				cr.status.deactivate(failureType{source: t.String(), action: "dial"}, err.Error())
			}
		} else {
			cr.status.activate()
			if cr.lg != nil {
				cr.lg.Info(
					"established TCP streaming connection with remote peer",
					zap.String("stream-reader-type", cr.typ.String()),
					zap.String("local-member-id", cr.tr.ID.String()),
					zap.String("remote-peer-id", cr.peerID.String()),
				)
			}
			// 如果未出现异常，则开始读取对端返回的消息，并将读取到的消息写入streamReader.recvc通道中
			err = cr.decodeLoop(rc, t)
			if cr.lg != nil {
				cr.lg.Warn(
					"lost TCP streaming connection with remote peer",
					zap.String("stream-reader-type", cr.typ.String()),
					zap.String("local-member-id", cr.tr.ID.String()),
					zap.String("remote-peer-id", cr.peerID.String()),
					zap.Error(err),
				)
			}
			switch {
			// 读取所有数据
			case err == io.EOF:
			// 远程
			case transport.IsClosedConnError(err):
			default:
				cr.status.deactivate(failureType{source: t.String(), action: "read"}, err.Error())
			}
		}
		// 关闭连接
		err = cr.rl.Wait(cr.ctx)
		if cr.ctx.Err() != nil {
			if cr.lg != nil {
				cr.lg.Info(
					"stopped stream reader with remote peer",
					zap.String("stream-reader-type", t.String()),
					zap.String("local-member-id", cr.tr.ID.String()),
					zap.String("remote-peer-id", cr.peerID.String()),
				)
			}
			close(cr.done)
			return
		}
		if err != nil {
			if cr.lg != nil {
				cr.lg.Warn(
					"rate limit on stream reader with remote peer",
					zap.String("stream-reader-type", t.String()),
					zap.String("local-member-id", cr.tr.ID.String()),
					zap.String("remote-peer-id", cr.peerID.String()),
					zap.Error(err),
				)
			}
		}
	}
}

// decodeLoop（）方法是streamReader中的核心方法，它会从底层的网络连接读取数据并进行反序列化，
// 之后将得到的消息实例写入reeve通道（或prope通道）中，等待Peer进行处理，
func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error {
	var dec decoder
	cr.mu.Lock()
	// 根据使用的协议版本创建对应的decoder实例，
	switch t {
	case streamTypeMsgAppV2:
		dec = newMsgAppV2Decoder(rc, cr.tr.ID, cr.peerID)
	case streamTypeMessage:
		dec = &messageDecoder{r: rc}	// messageDeeoder主妥负责从连接中读取数据
	default:
		if cr.lg != nil {
			cr.lg.Panic("unknown stream type", zap.String("type", t.String()))
		}
	}
	// 检测streamReader是否已经关闭了，若关闭则返回异常
	select {
	case <-cr.ctx.Done():
		cr.mu.Unlock()
		if err := rc.Close(); err != nil {
			return err
		}
		return io.EOF
	default:
		cr.closer = rc
	}
	cr.mu.Unlock()

	for {
		// ／从底层连接中读取数据，并反序列化成raftpb.Message实例
		m, err := dec.decode()
		if err != nil {
			cr.mu.Lock()
			cr.close()
			cr.mu.Unlock()
			return err
		}

		// gofail go:var raftDropHeartbeat struct{}
		// continue labelRaftDropHeartbeat 
		receivedBytes.WithLabelValues(types.ID(m.From).String()).Add(float64(m.Size()))

		cr.mu.Lock()
		paused := cr.paused
		cr.mu.Unlock()
		// 检测是否已经暂停从该连接上读取消息
		if paused {
			continue
		}
		// 忽略连接层的心跳消息，注意与Raft的心跳消息进行区分
		if isLinkHeartbeatMessage(&m) {
			continue
		}
		// 根据读取到的消息类型，选择对应通道进行写入
		recvc := cr.recvc
		if m.Type == raftpb.MsgProp {
			recvc = cr.propc
		}

		select {
		// 将消息写入对应的远远中，之后会交给底层的Raft状态机进行处理
		case recvc <- m:
		// recvc通过满了之后，只能丢弃消息，并打印日志
		default:
			if cr.status.isActive() {
				if cr.lg != nil {
					cr.lg.Warn(
						"dropped internal Raft message since receiving buffer is full (overloaded network)",
						zap.String("message-type", m.Type.String()),
						zap.String("local-member-id", cr.tr.ID.String()),
						zap.String("from", types.ID(m.From).String()),
						zap.String("remote-peer-id", types.ID(m.To).String()),
						zap.Bool("remote-peer-active", cr.status.isActive()),
					)
				}
			} else {
				if cr.lg != nil {
					cr.lg.Warn(
						"dropped Raft message since receiving buffer is full (overloaded network)",
						zap.String("message-type", m.Type.String()),
						zap.String("local-member-id", cr.tr.ID.String()),
						zap.String("from", types.ID(m.From).String()),
						zap.String("remote-peer-id", types.ID(m.To).String()),
						zap.Bool("remote-peer-active", cr.status.isActive()),
					)
				}
			}
			recvFailures.WithLabelValues(types.ID(m.From).String()).Inc()
		}
	}
}

func (cr *streamReader) stop() {
	cr.mu.Lock()
	cr.cancel()
	cr.close()
	cr.mu.Unlock()
	<-cr.done
}

// streamReader.dial（）方法主要负责与对端节点建立连接，其中包含了多种异常情况的处理，
func (cr *streamReader) dial(t streamType) (io.ReadCloser, error) {
	u := cr.picker.pick()	// 获取对端节点暴露的一个URL
	uu := u
	// 根据使用协议的版本和节点ID创建最终的URL地址
	uu.Path = path.Join(t.endpoint(cr.lg), cr.tr.ID.String())

	if cr.lg != nil {
		cr.lg.Debug(
			"dial stream reader",
			zap.String("from", cr.tr.ID.String()),
			zap.String("to", cr.peerID.String()),
			zap.String("address", uu.String()),
		)
	}
	// 创建一个GET请求
	req, err := http.NewRequest("GET", uu.String(), nil)
	if err != nil {
		cr.picker.unreachable(u)
		return nil, fmt.Errorf("failed to make http request to %v (%v)", u, err)
	}
	// 设置HTTP请求头
	req.Header.Set("X-Server-From", cr.tr.ID.String())
	req.Header.Set("X-Server-Version", version.Version)
	req.Header.Set("X-Min-Cluster-Version", version.MinClusterVersion)
	req.Header.Set("X-Etcd-Cluster-ID", cr.tr.ClusterID.String())
	req.Header.Set("X-Raft-To", cr.peerID.String())
	// 将当前节点暴露的URL也一起发送给对端节点
	setPeerURLsHeader(req, cr.tr.URLs)

	req = req.WithContext(cr.ctx)

	cr.mu.Lock()
	select {
	case <-cr.ctx.Done():
		cr.mu.Unlock()
		return nil, fmt.Errorf("stream reader is stopped")
	default:
	}
	cr.mu.Unlock()

	resp, err := cr.tr.streamRt.RoundTrip(req)	// 发送请求，ping-pong
	if err != nil {
		cr.picker.unreachable(u)
		return nil, err
	}
	// 检测版本信息
	rv := serverVersion(resp.Header)
	lv := semver.Must(semver.NewVersion(version.Version))
	if compareMajorMinorVersion(rv, lv) == -1 && !checkStreamSupport(rv, t) {
		httputil.GracefulClose(resp)
		cr.picker.unreachable(u)
		return nil, errUnsupportedStreamType
	}

	switch resp.StatusCode {	// 根据HTTP的响应码进行处理
	case http.StatusGone:
		httputil.GracefulClose(resp)
		cr.picker.unreachable(u)
		reportCriticalError(errMemberRemoved, cr.errorc)
		return nil, errMemberRemoved

	case http.StatusOK:
		return resp.Body, nil

	case http.StatusNotFound:
		httputil.GracefulClose(resp)
		cr.picker.unreachable(u)
		return nil, fmt.Errorf("peer %s failed to find local node %s", cr.peerID, cr.tr.ID)

	case http.StatusPreconditionFailed:
		b, err := io.ReadAll(resp.Body)
		if err != nil {
			cr.picker.unreachable(u)
			return nil, err
		}
		httputil.GracefulClose(resp)
		cr.picker.unreachable(u)

		switch strings.TrimSuffix(string(b), "\n") {
		case errIncompatibleVersion.Error():
			if cr.lg != nil {
				cr.lg.Warn(
					"request sent was ignored by remote peer due to server version incompatibility",
					zap.String("local-member-id", cr.tr.ID.String()),
					zap.String("remote-peer-id", cr.peerID.String()),
					zap.Error(errIncompatibleVersion),
				)
			}
			return nil, errIncompatibleVersion

		case errClusterIDMismatch.Error():
			if cr.lg != nil {
				cr.lg.Warn(
					"request sent was ignored by remote peer due to cluster ID mismatch",
					zap.String("remote-peer-id", cr.peerID.String()),
					zap.String("remote-peer-cluster-id", resp.Header.Get("X-Etcd-Cluster-ID")),
					zap.String("local-member-id", cr.tr.ID.String()),
					zap.String("local-member-cluster-id", cr.tr.ClusterID.String()),
					zap.Error(errClusterIDMismatch),
				)
			}
			return nil, errClusterIDMismatch

		default:
			return nil, fmt.Errorf("unhandled error %q when precondition failed", string(b))
		}

	default:
		httputil.GracefulClose(resp)
		cr.picker.unreachable(u)
		return nil, fmt.Errorf("unhandled http status %d", resp.StatusCode)
	}
}

func (cr *streamReader) close() {
	if cr.closer != nil {
		if err := cr.closer.Close(); err != nil {
			if cr.lg != nil {
				cr.lg.Warn(
					"failed to close remote peer connection",
					zap.String("local-member-id", cr.tr.ID.String()),
					zap.String("remote-peer-id", cr.peerID.String()),
					zap.Error(err),
				)
			}
		}
	}
	cr.closer = nil
}

func (cr *streamReader) pause() {
	cr.mu.Lock()
	defer cr.mu.Unlock()
	cr.paused = true
}

func (cr *streamReader) resume() {
	cr.mu.Lock()
	defer cr.mu.Unlock()
	cr.paused = false
}

// checkStreamSupport检查给定版本中是否支持流类型。
func checkStreamSupport(v *semver.Version, t streamType) bool {
	nv := &semver.Version{Major: v.Major, Minor: v.Minor}
	for _, s := range supportedStream[nv.String()] {
		if s == t {
			return true
		}
	}
	return false
}
